/*******************************************************************************
 ** Name: ThreadMgr.cpp                                          **
 ** Description:                                                              **
 **                                                                           **
 ** Open Source Initiative (OSI) Approved License                             **
 **                                                                           **
 **  Copyright (c) 2009-2010 Barry "Gian" James  <bjames@munixos.net>         **
 **  All rights reserved.                                                     **
 **                                                                           **
 **  Redistribution and use in source and binary forms, with or without       **
 **  modification, are permitted provided that the following conditions are   **
 **  met:                                                                     **
 **                                                                           **
 **  * Redistributions of source code must retain the above copyright notice, **
 **    this list of conditions and the following disclaimer.                  **
 **  * Redistributions in binary form must reproduce the above copyright      **
 **    notice, this list of conditions and the following disclaimer in the    **
 **    documentation and/or other materials provided with the distribution.   **
 **                                                                           **
 **  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS      **
 ** "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED **
 ** TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A           **
 ** PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT        **
 ** HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,    **
 ** SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED  **
 ** TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR    **
 ** PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF    **
 ** LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING      **
 ** NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS        **
 ** SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.              **
 **                                                                           **
 ******************************************************************************/
// Created on March 24, 2010, 3:21 AM
// Modified $Date$ by $Author$
#define  ThreadMgr_cpp_ID    "$Id$"
extern "C" {
#include <assert.h>
#include <unistd.h>
}
#include <string.h>

#include "threadmgr.hpp"
#include "log.hpp"

void * checkThreadQueue(void * vtmgr);

void * checkThreadQueue(void * vtmgr)
{
    DBGFUNC();

    ThreadMgr * tmgr = reinterpret_cast<ThreadMgr*>(vtmgr);
    ThreadPoolBlock *   workMsg;
    int threadID = tmgr->_assignID();
    debug::log->out(LOG_DEBUG, "Thread #%d spinning up",threadID);

    pthread_mutex_t * mutex = tmgr->QueueMutex();

    for (;;)
    {
        pthread_mutex_lock(mutex);
        while(tmgr->QueueIsEmpty()) {
            debug::log->out(LOG_DEBUG,"Thread(%d): QueueIsEmpty()",threadID);
            pthread_cond_wait(tmgr->QueueNotEmptyCond(),mutex);
        }

        debug::log->out(LOG_DEBUG,"Thread(%d) Processing msg queue", threadID);
        workMsg = tmgr->NextMsg();

        pthread_mutex_unlock(mutex);

        (*(workMsg->func))(workMsg->arg);
        delete workMsg;
    }
    return 0;
}

//////////////////////////////[ CriticalSection
CriticalSection::CriticalSection()
{
	bIsValid = false;
	criticalSection = (pthread_mutex_t*)malloc(sizeof(pthread_mutex_t));
	if (criticalSection != NULL)
		if (pthread_mutex_init(criticalSection,NULL) == 0)
			bIsValid = true;
}

CriticalSection::~CriticalSection()
{
	if (criticalSection != NULL)
	{
		pthread_mutex_destroy(criticalSection);
		free(criticalSection);
	}
	bIsValid = false;
}

bool
CriticalSection::Enter()
{
	assert(bIsValid);
	bool output = false;
	if (pthread_mutex_lock(criticalSection) == 0)
		output = true;
	return output;
}

bool
CriticalSection::Leave()
{
	assert(bIsValid);
	bool output = false;
	if (pthread_mutex_unlock(criticalSection) == 0)
		output = true;
	return output;
}

//////////////////////////////[ ThreadMgr
int
ThreadMgr::idx = 0;

pthread_mutex_t
ThreadMgr::readSetMutex = PTHREAD_MUTEX_INITIALIZER;

pthread_mutex_t
ThreadMgr::writeSetMutex = PTHREAD_MUTEX_INITIALIZER;

bool
ThreadMgr::Init()
{
    DBGFUNC();

    int ret;
    pthread_condattr_t  pattr;

    threads = new pthread_t [maxThreads];

    if ( (ret = pthread_mutex_init(&lockQueue,NULL)) != 0) {
        debug::log->out("%s pthread_mutex_init() %s",__PRETTY_FUNCTION__,strerror(ret));
        return false;
    }
    if ( (ret = pthread_condattr_init(&pattr)) != 0) {
        debug::log->out("pthread_condattr_init: %s", strerror(ret));
        return false;
    }
    if (pthread_condattr_setpshared(&pattr,PTHREAD_PROCESS_SHARED) != 0) {
        debug::log->out("%s pthread_condattr_setpshared");
        return false;
    }
    if ( (ret = pthread_cond_init(&condQueueNotEmpty,&pattr)) != 0) {
        debug::log->out("%s pthread_cond_init() %s",__PRETTY_FUNCTION__,strerror(ret));
        return false;
    }

    for (int i = 0; i != maxThreads; i++) {
        if ( (ret = pthread_create(&threads[i],NULL,checkThreadQueue,(void*)this)) != 0) {
            debug::log->out("ERROR>> pthread_create() %s",strerror(ret));
            return false;
        }
    }

    pthread_condattr_getpshared(&pattr,&ret);
    if (ret == PTHREAD_PROCESS_SHARED)
        debug::log->out(LOG_DEBUG,"PTHREAD_PROCESS_SHARED");
    else
        debug::log->out(LOG_DEBUG,"%d",ret);

    return true;
}

int
ThreadMgr::AssignWork(tfunc f, void* a)
{
    DBGFUNC();

    ThreadPoolBlock *   newBlock;

    pthread_mutex_lock(&lockQueue);

    newBlock = new ThreadPoolBlock();
    newBlock->func  = f;
    newBlock->arg   = a;

    msgQueue.push(newBlock);
    pthread_cond_signal(&condQueueNotEmpty);

    pthread_mutex_unlock(&lockQueue);
    
    return msgQueue.size();
}

ThreadMgr::ThreadMgr(int m)
{
    maxThreads = m;
    threads    = NULL;
}


ThreadMgr::~ThreadMgr()
{

}

